package com.gome.ocean.controller.datax;

import java.lang.reflect.InvocationTargetException;
import java.net.URL;
import java.util.Date;
import java.util.Enumeration;
import java.util.List;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.commons.beanutils.BeanUtilsBean;
import org.apache.commons.beanutils.ConvertUtilsBean;
import org.apache.commons.beanutils.PropertyUtilsBean;
import org.apache.commons.lang.StringUtils;
import org.apache.http.client.methods.HttpGet;
import org.quartz.SchedulerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.ModelMap;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;

import com.alibaba.fastjson.JSONObject;
import com.github.pagehelper.Page;
import com.gome.ocean.common.constants.DataxContants;
import com.gome.ocean.common.enums.State;
import com.gome.ocean.common.exception.DataxRemoteInvokeException;
import com.gome.ocean.common.exception.JobServiceException;
import com.gome.ocean.common.utils.AddressUtils;
import com.gome.ocean.common.utils.DateConverter;
import com.gome.ocean.common.utils.HttpClientUtil;
import com.gome.ocean.common.utils.ResponsePage;
import com.gome.ocean.dao.model.authority.ProjectBO;
import com.gome.ocean.dao.model.datax.CreateJobVO;
import com.gome.ocean.dao.model.datax.JobBO;
import com.gome.ocean.dao.model.datax.JobHistoryBO;
import com.gome.ocean.service.authority.ProjectService;
import com.gome.ocean.service.base.AuthorityBaseController;
import com.gome.ocean.service.datax.JobHistoryService;
import com.gome.ocean.service.datax.JobService;
import com.gome.ocean.service.task.CallBack;
import com.gome.ocean.service.thread.ThreadFactory;
import com.gome.ocean.service.utils.CmdToolkit;
import com.gome.ocean.service.utils.job.CreateJobUtils;

/**
 * Created by liuqingxia on 2016/1/14.
 */
@Controller
@RequestMapping(value = "/job")
public class JobController extends AuthorityBaseController {

    private static final Logger LOG = LoggerFactory.getLogger(JobController.class);

    @Autowired
    private JobService          jobService;
    @Autowired
    private JobHistoryService   jobHistoryService;
    @Autowired
    private ThreadFactory       threadFactory;
    @Autowired
    private ProjectService      projectService;

    @RequestMapping(value = "/insert", method = RequestMethod.GET)
    public String jobAddGet(ModelMap m) {
        try {
            m.put("projectNames", getProperProjects());
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            m.put("errorMsg", "查询项目组列表失败");
            return "job/insert";
        }
        return "job/insert";
    }

    @RequestMapping(value = "/insert", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage jobAddPost(HttpServletRequest req, HttpServletResponse res) {
        CreateJobVO createJobVO;
        ResponsePage page = getResponsePage();
        try {
            createJobVO = CreateJobUtils.getJobVO(req, super.getIpAddr(req));
            boolean flag = super.permitAccess(createJobVO);
            if (flag) {
                jobService.insertJob(createJobVO);
                return page.setSuccessMsg("任务保存成功");
            }
            return page.setErrorMsg("您没有项目组权限");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg(e.getMessage());
        }
    }

    @RequestMapping(value = "/remove", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage removeJob(HttpServletRequest req, @RequestParam("id") long id) {
        ResponsePage page = getResponsePage();
        String email = getCurrentUser(req);
        try {
            JobBO job = jobService.findJob(id);
            if (null == job) {
                return page.setErrorMsg("数据库中不存在该条任务记录，ID=" + id);
            }
            Long projectId = job.getProjectId();
            ProjectBO project = projectService.findProjectById(projectId);
            if(project == null){
                return page.setErrorMsg("项目组未找到！");
            }
            job.setProject(project.getProjectName());
            boolean flag = super.permitAccess(job);
            if (flag) {
                if (jobService.removeJob(job)) {
                    LOG.debug("用户：" + email + ",删除id=" + id + "的任务");
                    return page.setSuccessMsg("删除任务成功");
                } else {
                    return page.setErrorMsg("删除任务失败");
                }
            }
            return page.setErrorMsg("用户没有项目组权限");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg("删除任务过程中出现异常，异常信息为" + e.getMessage());
        }
    }

    @RequestMapping(value = "/update", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage updateJob(@RequestParam("id") long id) {
        ResponsePage page = getResponsePage();
        String email = getCurrentUser();
        JobBO job;
        try {
            job = getJobBOFromRequest(getHttpRequest());
            if (null == job) {
                return page.setErrorMsg("获取任务参数失败");
            }
            job.setProject(job.getPipelineName());
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg("获取任务参数时发生异常，异常信息为：" + e.getMessage());
        }
        try {
            boolean flag = super.permitAccess(job);
            if (flag) {
                if (jobService.updateJob(job)) {
                    LOG.debug("用户：" + email + ",更新id=" + id + "的任务");
                    return page.setSuccessMsg("任务修改成功");
                } else {
                    return page.setErrorMsg("任务修改失败，请检查任务参数并稍后重试");
                }
            }
            return page.setErrorMsg("您没有项目组权限");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg("任务修改过程中发生异常，异常信息为：" + e.getMessage());
        }
    }

    private JobBO getJobBOFromRequest(HttpServletRequest req) throws InvocationTargetException, IllegalAccessException {
        JobBO job = new JobBO();
        ConvertUtilsBean convertUtils = new ConvertUtilsBean();
        DateConverter dateConverter = new DateConverter();
        convertUtils.register(dateConverter, Date.class);
        BeanUtilsBean beanUtils = new BeanUtilsBean(convertUtils, new PropertyUtilsBean());
        Enumeration names = req.getParameterNames();
        while (names.hasMoreElements()) {
            String name = (String) names.nextElement();
            beanUtils.setProperty(job, name, req.getParameter(name));
        }
        return job;
    }

    @RequestMapping(value = "/find", method = RequestMethod.GET)
    public String jobFindGet() {
        return "job/find";
    }

    @RequestMapping(value = "/findById", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage<JobBO> jobFindById(@RequestParam("id") long id) {
        ResponsePage<JobBO> page = getResponsePage();
        try {
            JobBO job = jobService.findJob(id);
            if (job == null) {
                return page.setErrorMsg("未查询到任务，id=" + id);
            }
            job.setProject(job.getPipelineName());
            boolean flag = super.permitAccess(job);
            if (flag) {
                return page.setSingleObject(job, "查询成功");
            }
            return page.setErrorMsg("您没有项目组权限");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg(e.getMessage());
        }
    }

    @RequestMapping(value = "/find", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage<JobBO> getJobs(@RequestParam("introduce") String introduce) {
        ResponsePage<JobBO> page = new ResponsePage();
        try {
            if (StringUtils.isBlank(introduce)) {
                return findJobsWithOutIntroduceParam();
            }
            return findJobsWithIntroduceParam(introduce);
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg(e.getMessage());
        }
    }

    /**
     * 根据introduce参数查询job
     *
     * @param introduce
     * @return
     */
    private ResponsePage findJobsWithIntroduceParam(String introduce) {
        ResponsePage<JobBO> page = getResponsePage();
        Page<JobBO> jobBOs;
        int pageNum = getPageNum();
        int pageSize = getPageSize();
        try {
            if (isAdmin()) {
                jobBOs = jobService.findJobsByIntroduce(introduce, pageNum, pageSize);
            } else {
                List<String> projectNames = getCurrentUserProjects();
                if (null == projectNames || projectNames.isEmpty()) {
                    return page.setErrorMsg("您还没有加入任何项目组，请联系管理员");
                }
                jobBOs = jobService.findJobsByIntroduceAndProject(introduce, projectNames, pageNum, pageSize);
            }
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg(ERROR_MSG);
        }
        if (null == jobBOs || jobBOs.isEmpty()) {
            return page.emptyPage("未查询到结果");
        }
        return page.setPage(jobBOs, "查询成功");
    }

    /**
     * 不带参数查询job
     *
     * @return
     */
    private ResponsePage findJobsWithOutIntroduceParam() {
        ResponsePage<JobBO> page = getResponsePage();
        Page<JobBO> jobBOs;
        int pageNum = getPageNum();
        int pageSize = getPageSize();
        try {
            if (isAdmin()) {
                jobBOs = jobService.findAllJobs(pageNum, pageSize);
            } else {
                List<String> projectNames = getCurrentUserProjects();
                if (null == projectNames || projectNames.isEmpty()) {
                    return page.setErrorMsg("您还没有加入任何项目组，请联系管理员");
                }
                jobBOs = jobService.findJobsByProjectNames(projectNames, pageNum, pageSize);
            }
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg(ERROR_MSG);
        }
        if (null == jobBOs || jobBOs.isEmpty()) {
            return page.emptyPage("未查询到结果");
        }
        return page.setPage(jobBOs, "查询成功");
    }

    @RequestMapping(value = "/findByProjectName", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage findJobsByProjectName(@RequestParam("projectName") String projectName, HttpServletRequest req) {
        int pageNum = getPageNumFromRequest(req);
        int pageSize = getPageSizeFromRequest(req);
        ResponsePage page = getResponsePage();
        Page jobBOs = null;
        try {
            if (isAdmin(req)) {
                jobBOs = jobService.findJobByProjectName(projectName, pageNum, pageSize);
            } else {
                List<String> projectNames = getCurrentUserProjects(req);
                if (!projectNames.contains(projectName)) {
                    return page.setErrorMsg("您只能查询您所在的组");
                }
                jobBOs = jobService.findJobByProjectName(projectName, pageNum, pageSize);
            }
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg(e.getMessage());
        }
        if (null == jobBOs || jobBOs.isEmpty()) {
            return page.emptyPage("未查询到结果");
        }
        return page.setPage(jobBOs, "查询成功");
    }

    @RequestMapping(value = "/findByExecuteType", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage findJobsByExecuteType(@RequestParam("executeType") int type) {
        int pageNum = getPageNum();
        int pageSize = getPageSize();
        ResponsePage page = getResponsePage();
        Page jobBOs;
        try {
            if (isAdmin()) {
                jobBOs = jobService.findJobsByExecuteType(type, pageNum, pageSize);
            } else {
                List<String> projectNames = getCurrentUserProjects();
                jobBOs = jobService.findJobsByExecuteTypeAndProjectNames(type, projectNames, pageNum, pageSize);
            }
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg(e.getMessage());
        }
        if (null == jobBOs || jobBOs.isEmpty()) {
            return page.emptyPage("未查询到结果");
        }
        return page.setPage(jobBOs, "查询成功");
    }

    @RequestMapping(value = "/start", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage startJob(HttpServletRequest req, @RequestParam("id") long id) {
        ResponsePage page = getResponsePage();
        String email = getCurrentUser(req);
        try {
            JobBO job = jobService.findJob(id);
            if (null == job) {
                return page.setErrorMsg("数据库中不存在该条任务记录，ID=" + id);
            }
            job.setProject(job.getPipelineName());
            boolean flag = super.permitAccess(job);
            if (flag) {
                if (jobService.startJob(job)) {
                    LOG.debug("用户：" + email + ",启动id=" + id + "的任务");
                    return page.setSuccessMsg("任务提交成功");
                } else {
                    return page.setErrorMsg("任务提交失败");
                }
            }
            return page.setErrorMsg("您没有项目权限");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg("任务执行过程中发生异常，异常信息为：" + e.getMessage());
        }
    }

    @RequestMapping(value = "/pause", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage pauseJob(HttpServletRequest req, @RequestParam("id") long id) {
        ResponsePage page = getResponsePage();
        String email = getCurrentUser(req);
        try {
            JobBO job = jobService.findJob(id);
            if (null == job) {
                return page.setErrorMsg("数据库中不存在该条任务记录，ID=" + id);
            }
            job.setProject(job.getPipelineName());
            boolean flag = super.permitAccess(job);
            if (flag) {
                if (jobService.pauseJob(job)) {
                    LOG.debug("用户：" + email + ",暂停id=" + id + "的任务");
                    return page.setSuccessMsg("该任务已经暂停");
                } else {
                    return page.setErrorMsg("该任务无法暂停");
                }
            }
            return page.setErrorMsg("您没有项目权限");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg("暂停任务过程中出现异常，异常信息为：" + e.getMessage());
        }
    }

    @RequestMapping(value = "/trigger", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage triggerJob(@RequestParam("id") long id) {
        ResponsePage page = getResponsePage();
        String email = getCurrentUser();
        try {
            JobBO job = jobService.findJob(id);
            if (null == job) {
                return page.setErrorMsg("数据库中不存在该条任务记录，ID=" + id);
            }
            job.setProject(job.getPipelineName());
            boolean flag = super.permitAccess(job);
            if (flag) {
                if (jobService.triggerJob(job)) {
                    LOG.info("用户：" + email + ",触发id=" + id + "的任务");
                    return page.setSuccessMsg("该任务已经触发");
                } else {
                    return page.setErrorMsg("该任务无法触发");
                }
            }
            return page.setErrorMsg("您没有项目组权限");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg("触发任务过程中出现异常，异常信息为：" + e.getMessage());
        }
    }

    @RequestMapping(value = "/resume", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage resumeJob(HttpServletRequest req, @RequestParam("id") long id) {
        ResponsePage page = getResponsePage();
        String email = getCurrentUser(req);
        try {
            JobBO job = jobService.findJob(id);
            if (null == job) {
                return page.setErrorMsg("数据库中不存在该条任务记录，ID=" + id);
            }
            job.setProject(job.getPipelineName());
            boolean flag = super.permitAccess(job);
            if (flag) {
                if (jobService.resumeJob(job)) {
                    LOG.debug("用户：" + email + ",恢复id=" + id + "的任务");
                    return page.setSuccessMsg("任务重新启动成功");
                } else {
                    return page.setErrorMsg("任务重新启动失败");
                }
            }
            return page.setErrorMsg("您没有项目组权限");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg("任务重新启动过程中出现异常，异常信息为" + e.getMessage());
        }
    }

    @RequestMapping(value = "/interrupt", method = RequestMethod.POST)
    @ResponseBody
    public ResponsePage interruptJob(@RequestParam("id") long id, HttpServletRequest req) {
        ResponsePage page = getResponsePage();
        try {
            final JobBO job = jobService.findJob(id);
            if (null == job) {
                return page.setErrorMsg("数据库中不存在此条记录，id = " + id);
            }
            job.setProject(job.getPipelineName());
            boolean flag = super.permitAccess(job);
            if (flag) {
                final JobHistoryBO jobHistoryBO = jobHistoryService.findCurrentJob(job.getId());
                if (jobHistoryBO == null) {
                    return page.setErrorMsg("该任务已经终止!");
                }
                if (jobHistoryBO.getExecuteIp().indexOf(AddressUtils.getHostIp()) != -1) {
                    if (job.isCronJob()) {
                        try {
                            jobService.interrupt(job);
                        } catch (SchedulerException | JobServiceException e) {
                            LOG.info("jobBO stop is error!", e);
                            return page.setErrorMsg("任务停止失败!jobId=" + id);
                        }
                    } else {
                        CmdToolkit.executeCmd(DataxContants.getKillTaskCommand(jobHistoryBO.getTJobId()),
                            new CallBack() {

                                @Override
                                public void callBack() {
                                    threadFactory.stop(jobHistoryBO.getTJobId());

                                    job.setState(State.FAIL.getValue());
                                    job.setEndTime(new Date());

                                    jobHistoryBO.setState(job.getState());
                                    jobHistoryBO.setSubmitUser(job.getSubmitUser());
                                    jobHistoryBO.setSubmitTime(job.getSubmitTime());
                                    jobHistoryBO.setSubmitIp(job.getSubmitIp());
                                    jobHistoryBO.setPipelineName(job.getPipelineName());
                                    jobHistoryBO.setCronExpression(job.getCronExpression());

                                    jobHistoryBO.setStartTime(job.getStartTime());
                                    jobHistoryBO.setEndTime(job.getEndTime());
                                    jobHistoryBO.setSpeedBytes("0");
                                    jobHistoryBO.setSpeedRecords(0l);
                                    jobHistoryBO.setTotalRecords(0l);
                                    jobHistoryBO.setErrorRecords(0l);
                                    jobService.updateJobBO(job);
                                    jobHistoryService.updateJobHistoryById(jobHistoryBO);
                                }
                            });
                        return page.setSuccessMsg("任务终止成功!");
                    }
                } else {
                    if (StringUtils.isNotEmpty(jobHistoryBO.getExecuteIp())) {
                        String interruptPath = DataxContants.getDataxInterruptPath(jobHistoryBO.getExecuteIp(),
                            jobHistoryBO.getTJobId());
                        HttpClientUtil httpClientUtil = new HttpClientUtil();
                        try {
                            URL url = new URL(interruptPath);
                            HttpGet httpGet = HttpClientUtil.getGetRequest();
                            httpGet.setURI(url.toURI());
                            String result = httpClientUtil.executeAndGetWithFailedRetry(httpGet, 6, 1000l);
                            if (StringUtils.isNotEmpty(result)) {
                                result = JSONObject.parseObject(result).getString("result");
                                if (result.equals("ok")) {
                                    return page.setSuccessMsg("任务终止成功!");
                                } else {
                                    return page.setErrorMsg("任务终止失败!");
                                }
                            }
                            return page.setErrorMsg("任务终止失败!");
                        } catch (Exception e) {
                            throw new DataxRemoteInvokeException("终止任务失败！:" + interruptPath, e);
                        }
                    }
                }
            }
            return page.setErrorMsg("您没有项目组权限");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            return page.setErrorMsg(e.getMessage());
        }
    }
}
